package com.scheduler.actor;

import com.alibaba.fastjson.JSON;
import com.google.common.cache.Cache;
import com.google.common.cache.CacheBuilder;
import com.scheduler.actor.exception.WorkActorInvocationException;
import com.scheduler.actor.exception.WorkActorRemoteExecutionException;
import com.scheduler.actor.server.VertxServer;
import com.scheduler.actor.strategy.RoundRobinWithoutLocalhostRouterStrategy;
import com.scheduler.actor.strategy.RouterStrategy;
import com.scheduler.servicediscover.ServiceDiscoverHeartbeat;
import com.scheduler.servicediscover.Worker;
import com.scheduler.utils.okhttp.OkHttpClient;
import com.scheduler.utils.okhttp.ThrowableUtil;
import com.scheduler.utils.trace.TraceUtil;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;

import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Function;
import java.util.stream.Collectors;

/**
 * 可以创建多个实例，监听不同端口即可。
 * 1. 默认提供的是基于配置中心的组网和集群内调度
 * 2. 也可以提供单独的监听作为服务端，异构系统进行客户端调用，方便聚合接口
 */
@Slf4j
public class WorkerActorEventBus {

    public static final String OK = "OK";
    public static final String FAIL = "FAIL";
    public static final String NO_AUTH = "NO_AUTH";
    public static final String NO_PROVIDER = "NO_PROVIDER";
    public static final String ACCEPT = "text/plain; charset=utf-8";

    private static final String DEFAULT_RPC_URL_PREFIX = "http://%s:%s";
    private static final String DEFAULT_RPC_URL = "/eventBus/dispatch";
    private static final String DEFAULT_AUTH = "DE8D65119568C74EA8BF1C73";
    private static final int DEFAULT_PORT = 7080;

    public static final Map<Integer, WorkerActorEventBus> workerActorEventBusHolder = new ConcurrentHashMap<>();

    public static final WorkerActorEventBus defaultWorkerActorEventBus = new WorkerActorEventBus();
    /**
     * 每个实例只能初始化一次
     */
    public final AtomicBoolean initialized = new AtomicBoolean(false);

    /**
     * workers暂时不用，直接通过服务发现的功能查询即可{@link ServiceDiscoverHeartbeat#getActiveServerList()}
     */
    private final List<Worker> workers = null;
    /**
     * 实际工作的 Actor ，用于任务分发和回调
     */
    private final Map<String, WorkerActor> workerActorMap = new ConcurrentHashMap<>();

    /**
     * 分发任务URL
     */
    @Getter
    @Setter
    private String rpcUrlPrefix;
    @Getter
    @Setter
    private String rpcUrl;
    @Getter
    @Setter
    private String auth;
    @Getter
    @Setter
    private int port;

    private WorkerActorEventBus() {
        this(DEFAULT_PORT);
    }

    public WorkerActorEventBus(int port) {
        this(DEFAULT_AUTH, port);
    }

    public WorkerActorEventBus(String auth, int port) {
        this(DEFAULT_RPC_URL, auth, port);
    }

    public WorkerActorEventBus(String rpcUrl, String auth, int port) {
        this(DEFAULT_RPC_URL_PREFIX, rpcUrl, auth, port);
    }

    public WorkerActorEventBus(String rpcUrlPrefix, String rpcUrl, String auth, int port) {
        this.rpcUrlPrefix = rpcUrlPrefix;
        this.rpcUrl = rpcUrl;
        this.auth = auth;
        this.port = port;
        // 端口不能重复，重复直接报错
        workerActorEventBusHolder.compute(port, (port1, existWorkerActorEventBus) -> {
            if (existWorkerActorEventBus != null) {
                throw new IllegalArgumentException("http listener port duplicate " + port1);
            } else {
                return this;
            }
        });
    }

    public void start(int port) {
        this.port=port;
        if (initialized.compareAndSet(false, true)) {
            VertxServer vertxServer = new VertxServer(this);
            vertxServer.startServer(port);
        }
    }

    /**
     * 注册回调WorkerActor
     */
    public synchronized void registerCallbackWorkActor(WorkerActor workerActor) {
        Objects.requireNonNull(workerActor);
        Objects.requireNonNull(workerActor.getName());
        Objects.requireNonNull(workerActor.getDispatcherMap());
        String workerActorName = workerActor.getName();
        boolean containsKey = workerActorMap.containsKey(workerActorName);
        if (containsKey) {
            throw new IllegalArgumentException("重复定义的workerActor.name: " + workerActorName);
        }
        if (workerActor.getDispatcherMap().isEmpty()) {
            throw new IllegalArgumentException("没有配置DispatcherMap");
        }
        // 转为 Map
        workerActorMap.put(workerActorName, workerActor);
    }

    /**
     * 广播任务：对服务发现的集群所有 IP 地址广播
     */
    public Map<String, String> broadcastTask(ActorMessage actorMessage) {
        Map<String, String> resultMap = new HashMap<>();
        try {
            List<String> serverList = getServerList();
            for (String ip : serverList) {
                try {
                    actorMessage.setAuth(auth);
                    actorMessage.setTargetIp(ip);
                    actorMessage.setSourceIp(localhost);
                    String traceId = fillTraceId(actorMessage);

                    if (isLocalHost(ip)) {
                        // 本机则直接调用分发
                        Object result = receiptTask(actorMessage);
                        resultMap.put(ip, Objects.toString(result, null));
                        continue;
                    }
                    // 远程分发到别的机器
                    String url = String.format(rpcUrlPrefix, ip, port) + rpcUrl;
                    String result = OkHttpClient.doPost(url, JSON.toJSONString(actorMessage),
                            OkHttpClient.Header.of("sourceIp", localhost),
                            OkHttpClient.Header.of("traceId", traceId),
                            OkHttpClient.Header.of("accept", ACCEPT));
                    log.info("广播任务成功，targetIp={}, result={}", ip, result);
                    resultMap.put(ip, result);
                } catch (Throwable t) {
                    log.error("广播任务异常，targetIp={}", ip, t);
                    resultMap.put(ip, FAIL);
                }
            }
        } catch (Throwable t) {
            log.error("广播任务异常", t);
        }
        return resultMap;
    }

    /**
     * 分发任务出去或本机执行：对服务发现的集群所有 IP 地址轮询调用，包含本机
     */
    public Object dispatchTask(ActorMessage actorMessage) {
        String targetServerIp = null;
        try {
            targetServerIp = getTargetWorkerActorIp(actorMessage);

            actorMessage.setAuth(auth);
            actorMessage.setTargetIp(targetServerIp);
            actorMessage.setSourceIp(localhost);
            String traceId = fillTraceId(actorMessage);

            if (targetServerIp == null || isLocalHost(targetServerIp)) {
                // 本机则直接调用分发
                return receiptTask(actorMessage);
            }
            // 远程分发到别的机器
            String url = String.format(rpcUrlPrefix, targetServerIp, port) + rpcUrl;
            String result = OkHttpClient.doPost(url, JSON.toJSONString(actorMessage),
                    OkHttpClient.Header.of("sourceIp", localhost),
                    OkHttpClient.Header.of("traceId", traceId),
                    OkHttpClient.Header.of("accept", ACCEPT));
            log.info("分发任务成功，targetIp={}, result={}", targetServerIp, result);
            return result;
        } catch (Throwable t) {
            log.error("分发任务异常，targetIp={}", targetServerIp, t);
            return FAIL;
        }
    }

    /**
     * 远程调用：对指定的 HTTP 地址调用，需要在创建 WorkerActorEventBus 时指定固定 HTTP 地址。本功能无需服务发现支持。
     */
    public String invokeTask(ActorMessage actorMessage) {
        String url = null;
        try {
            actorMessage.setAuth(auth);
            actorMessage.setTimestamp(System.currentTimeMillis());
            actorMessage.setTargetIp("unknown");
            actorMessage.setSourceIp(localhost);
            String traceId = fillTraceId(actorMessage);
            // 签名
            actorMessage.doSignAndClearAuth();

            // 远程调用
            url = String.format(rpcUrlPrefix, port) + rpcUrl;
            String result = OkHttpClient.doPost(url, JSON.toJSONString(actorMessage),
                    OkHttpClient.Header.of("sourceIp", localhost),
                    OkHttpClient.Header.of("traceId", traceId),
                    OkHttpClient.Header.of("accept", ACCEPT));

            // 处理结果
            if (StringUtils.equalsAny(result, WorkerActorEventBus.NO_AUTH, WorkerActorEventBus.NO_PROVIDER)) {
                log.info("执行任务失败，result={}", result);
                throw new WorkActorRemoteExecutionException(result);
            }

            return result;
        } catch (WorkActorInvocationException t) {
            log.error("执行任务异常, url={}", url, t);
            throw t;
        } catch (Throwable t) {
            log.error("执行任务异常, url={}", url, t);
            throw new WorkActorInvocationException(t);
        }
    }

    /**
     * 接收并处理任务
     * 如有需要，在调用该方法前，写入具体的 MDC traceId 信息，以便日志和全链路跟踪使用
     */
    public Object receiptTask(ActorMessage actorMessage) {
        try {
            String actorName = actorMessage.getActorName();
            String actorDispatcher = actorMessage.getActorDispatcher();
            String auth = actorMessage.getAuth();
            String sign = actorMessage.getSign();

            if (sign != null) {
                // 优先用签名校验：写入自己的 auth，并重签名，对比签名是否正确
                actorMessage.setAuth(this.auth);
                actorMessage.doSignAndClearAuth();
                String newSign = actorMessage.getSign();
                if (!Objects.equals(newSign, sign)) {
                    log.warn("签名校验失败");
                    return NO_AUTH;
                }
            } else {
                if (!Objects.equals(auth, this.auth)) {
                    // 如果没有签名，则用auth认证
                    log.warn("无效请求");
                    return NO_AUTH;
                }
            }
            if (StringUtils.isAnyBlank(actorName, actorDispatcher)) {
                log.warn("无效请求，参数为空");
                return NO_PROVIDER;
            }

            if (workerActorMap.isEmpty()) {
                log.warn("无法处理请求，没有注册WorkActorList");
                return NO_PROVIDER;
            }
            WorkerActor workerActor = workerActorMap.get(actorName);
            if (workerActor == null) {
                log.warn("无法处理请求，没有找到注册WorkerActor，actorName={}", actorName);
                return NO_PROVIDER;
            }
            Map<String, Function<ActorMessage, ?>> dispatcherMap = workerActor.getDispatcherMap();
            Function<ActorMessage, ?> function = dispatcherMap.get(actorDispatcher);
            if (function == null) {
                log.warn("无法处理请求，没有找到注册dispatcher，actorName={}，dispatcher={}", actorName, actorDispatcher);
                return NO_PROVIDER;
            }

            if (ActorMessage.TaskType.MAP_SIMPLE.name().equals(actorMessage.getTaskType())) {
                ThreadPoolExecutor threadPoolExecutor = workerActor.getThreadPoolExecutor();
                if (threadPoolExecutor != null) {
                    // 提交到线程执行任务
                    try {
                        Runnable runnable = () -> function.apply(actorMessage);
                        Function<Runnable, Runnable> runnableWrapper = workerActor.getRunnableWrapper();
                        if (runnableWrapper != null) {
                            runnable = runnableWrapper.apply(runnable);
                        }
                        threadPoolExecutor.submit(runnable);
                    } catch (RejectedExecutionException e) {
                        log.error("线程拒绝处理", e);
                        return FAIL;
                    }
                    return OK;
                } else {
                    // 没有线程池则直接执行任务（注意，为了和线程池的任务能对上，此处的实现建议返回 OK 或者 FAIL，以保持和线程池的方式一致）
                    @SuppressWarnings("UnnecessaryLocalVariable")
                    Object result = function.apply(actorMessage);
                    return result;
                }
            } else if (ActorMessage.TaskType.SIMPLE.name().equals(actorMessage.getTaskType())) {
                // 没有线程池则直接执行任务
                @SuppressWarnings("UnnecessaryLocalVariable")
                Object result = function.apply(actorMessage);
                return result;
            } else {
                return NO_PROVIDER;
            }
        } catch (Throwable t) {
            log.info("触发任务异常", t);
            return FAIL + ":" + ThrowableUtil.toMessageStringRootCause(t);
        }
    }

    private String fillTraceId(ActorMessage actorMessage) {
        if (actorMessage.getContextInfo() == null) {
            actorMessage.setContextInfo(new HashMap<>());
        }
        // 如果有 traceId 就直接返回
        String traceId = (String) actorMessage.getContextInfo().get("traceId");
        if (traceId != null) {
            return traceId;
        }
        // 如果没有 traceId 就自己生成一个
        traceId = Optional.ofNullable(TraceUtil.getTraceId()).orElseGet(() -> "N_" + TraceUtil.getOrCreateTraceId(null));
        actorMessage.getContextInfo().put("traceId", traceId);
        return traceId;
    }

    private final Cache<String, List<Worker>> SERVER_LIST_CACHE = CacheBuilder.newBuilder().expireAfterWrite(5, TimeUnit.SECONDS).build();
    private final Cache<String, String> LOCAL_IP_CACHE = CacheBuilder.newBuilder().expireAfterWrite(5, TimeUnit.SECONDS).build();

    private String localhost;
    private final RouterStrategy defaultRouterStrategy = new RoundRobinWithoutLocalhostRouterStrategy();

    private boolean isLocalHost(String targetIp) {
        return Objects.equals(localhost, targetIp);
    }

    /**
     * targetWorkerActorIp 其实等价于 Worker#WorkerActor，此处简化一点。
     */
    private String getTargetWorkerActorIp(ActorMessage actorMessage) {
        try {
            List<Worker> serverList = SERVER_LIST_CACHE.get("L", () -> {
                if (localhost == null) {
                    localhost = LOCAL_IP_CACHE.get("LI", ServiceDiscoverHeartbeat.INSTANCE::getIp);
                }
                return ServiceDiscoverHeartbeat.INSTANCE.getActiveServerList();
            });

            // 根据actorMessage找到不同的路由规则进行路由，当前默认用轮询方式
            WorkerActor workerActor = workerActorMap.get(actorMessage.getActorName());
            if (workerActor != null && workerActor.getRouterStrategy() != null) {
                RouterStrategy routerStrategy = workerActor.getRouterStrategy();
                return Optional.ofNullable(routerStrategy.next(serverList, localhost)).map(Worker::getIpAddr).orElse(null);
            }
            // 默认逻辑
            return Optional.ofNullable(defaultRouterStrategy.next(serverList, localhost)).map(Worker::getIpAddr).orElse(null);
        } catch (ExecutionException e) {
            log.error("获取待分发任务服务器失败", e);
            return null;
        }
    }

    /**
     * 所有机器列表
     */
    private List<String> getServerList() {
        try {
            List<Worker> serverList = SERVER_LIST_CACHE.get("L", () -> {
                if (localhost == null) {
                    localhost = LOCAL_IP_CACHE.get("LI", ServiceDiscoverHeartbeat.INSTANCE::getIp);
                }
                return ServiceDiscoverHeartbeat.INSTANCE.getActiveServerList();
            });
            return serverList.stream().map(Worker::getIpAddr).collect(Collectors.toList());
        } catch (ExecutionException e) {
            log.error("获取待分发任务服务器失败", e);
            return Collections.emptyList();
        }
    }
}
